#include "threadpool.h"

#include <pthread.h>

const char *THREADPOOL_VERSION = "ThreadPool V1.1";

typedef struct _task_node {
    TaskFunc task;
    void *arg;
    struct _task_node *next;
} TaskNode;//任务节点

typedef struct {
    int                 ShutDown;//销毁标记

    pthread_mutex_t     Mutex;//锁
    pthread_cond_t      Cond;//条件变量

    int                 TaskNum;//等待中的任务数
    TaskNode*           TaskHead;//任务链表头
    TaskNode            _Task;//任务链表头结构体

    int                 MaxThreadNum;//最大任务线程数
    int                 TotalThreadNum;//目标线程数
    int                 ActiveThreadNum;//当前运行中的任务线程数
    int                 FreeThreadNum;//当前空闲线程数
    pthread_t*          Threads;//线程组
} ThreadPool_t;//句柄结构体

//================================================================== 
//Description: Get Version
//Return: Version String.
//==================================================================
const char *ThreadPool_Version(void)
{
    return THREADPOOL_VERSION;
}

//================================================================== 
//Description: Init One TaskNode
//Calls: 
//Calls By: ThreadPool_Add_Task_Tail/ThreadPool_Add_Task_Front
//Input：
//      task :   task function
//      arg  :   task arg /* If your task function dont need arg, you can just input NULL. */
//Return: Point of TaskNode, if init fail return NULL.
//==================================================================
static TaskNode *TaskNode_Init(TaskFunc task, void *arg)
{
    TaskNode *node = NULL;

    if(NULL == task) return NULL;

    node = (TaskNode *)malloc(sizeof(TaskNode));
    if(NULL == node) return NULL;

    node->next = NULL;
    node->task = task;
    node->arg = arg;

    return node;
}

//================================================================== 
//Description: Destory One TaskNode
//Calls:
//Calls By: routine/ThreadPool_Add_Task_Tail/ThreadPool_Add_Task_Front
//Input：
//      node    :   Point of TaskNode.
//Return: NON Return Value.
//==================================================================
static void TaskNode_Destory(TaskNode *node)
{
    if(NULL == node) return;
    node->next = NULL;
    free(node);
}

//================================================================== 
//Description: Push A TaskNode at List Tail
//Calls: 
//Calls By: ThreadPool_Add_Task_Tail
//Input：
//      head :   Point of list head.
//      node :   Point of TaskNode.
//Return: 
//      TP_SUCCESS
//      TP_NULL_POINT
//==================================================================
static int TaskNode_Push_Tail(TaskNode *head, TaskNode *node)
{
    TaskNode *p;
    if(NULL == head || NULL == node) return TP_NULL_POINT;

    p = head;
    while(NULL != p->next) {
        p = p->next;
    }

    p->next = node;

    return TP_SUCCESS;
}

//================================================================== 
//Description: Push A TaskNode at List Front
//Calls: 
//Calls By: ThreadPool_Add_Task_Front
//Input：
//      head :   Point of list head.
//      node :   Point of TaskNode.
//Return: 
//      TP_SUCCESS
//      TP_NULL_POINT
//==================================================================
static int TaskNode_Push_Front(TaskNode *head, TaskNode *node)
{
    if(NULL == head || NULL == node) return TP_NULL_POINT;

    node->next = head->next;
    head->next = node;

    return TP_SUCCESS;
}

//================================================================== 
//Description: Pop one TaskNode from list head
//Calls: 
//Calls By: routine
//Input：
//      head :   Point of lish head
//Return: Point of TaskNode, if get fail return NULL.
//==================================================================
static TaskNode *TaskNode_Pop(TaskNode *head)
{
    TaskNode *p = NULL;

    if(NULL == head) return NULL;

    //出链表
    p = head->next;
    head->next = p->next;
    p->next = NULL;

    return p;
}

//================================================================== 
//Description: PTHREAD_MUTEX_UNLOCK
//Calls: 
//Calls By: routine
//Input：
//      arg :   Point of pthread_mutex_t
//Return: NON
//==================================================================
static void handler(void *arg)
{
	pthread_mutex_unlock((pthread_mutex_t *)arg);
}

//================================================================== 
//Description: ThreadPool Work Thread Function
//Calls:
//Calls By: 
//Input：
//      arg :   Point of the ThreadPool
//Return: NULL
//==================================================================
static void *routine(void *arg)
{
    int index;
    TaskNode *task = NULL;
    ThreadPool_t *pool = (ThreadPool_t *)arg;

    if(NULL == pool) pthread_exit(NULL);

    while (1) {
        //解锁任务压栈
		pthread_cleanup_push(handler, (void *)&(pool->Mutex));
		pthread_mutex_lock(&(pool->Mutex));

        pool->FreeThreadNum++;//空闲线程加一

        //如果没有任务,且线程池未关闭
		while(pool->TaskNum == 0 && pool->ShutDown == 0) {
			pthread_cond_wait(&(pool->Cond), &(pool->Mutex));//用条件变量阻塞
            if(pool->TotalThreadNum < pool->ActiveThreadNum) {//判断是否需要退出当前线程
                break;
            }
		}

        //需要退出当前线程
        if(pool->TotalThreadNum < pool->ActiveThreadNum) {
            pool->ActiveThreadNum--;//活动线程数减少
            pool->FreeThreadNum--;//空闲线程减少
            for(index = 0; index < pool->MaxThreadNum; index++) {//找到线程组中的自己，清零
                if(pool->Threads[index] == pthread_self()) {
                    pool->Threads[index] = 0;
                    break;
                }
            }

            pthread_mutex_unlock(&(pool->Mutex));//解锁
            pthread_cond_signal(&(pool->Cond));//通知另一个线程取任务

            pthread_detach(pthread_self());//设置分离属性
            pthread_exit(NULL);//线程退出
        }

        //如果没有任务,且线程池关闭
		while(pool->TaskNum == 0 && pool->ShutDown == 1) {
			//解锁当前线程,并退出
			pthread_mutex_unlock(&(pool->Mutex));
			pthread_exit(NULL);
		}

        pool->FreeThreadNum--;//空闲线程减少

        //取出任务
        task = TaskNode_Pop(pool->TaskHead);
        if(NULL != task) pool->TaskNum--;//等待任务减少

        //释放线程锁,解锁函数弹出但不执行
		pthread_mutex_unlock(&(pool->Mutex));
		pthread_cleanup_pop(0);

        //如果取任务出错
        if(NULL == task) continue;

        //执行任务期间,线程不会被取消
		pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);
		(task->task)(task->arg);//执行任务
		pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);

        //销毁任务节点
        TaskNode_Destory(task);
        task = NULL;
    }
}

//================================================================== 
//Description: Insert one Work Thread into ThreadPool
//Calls: 
//Calls By: ThreadPool_Init/ThreadPool_Insert_Threads
//Input：
//      this    :   Point of the ThreadPool.
//Return:
//      TP_SUCCESS
//      TP_NULL_POINT
//      TP_MAX_THREADS
//      TP_INSERT_FAIL
//      TP_FAIL
//==================================================================
static int ThreadPool_Insert(ThreadPool_t *this)
{
    int index;

    if(NULL == this || NULL == this->Threads) return TP_NULL_POINT;//空指针
    if(this->ActiveThreadNum >= this->MaxThreadNum) return TP_MAX_THREADS;//已达上限

    for(index = 0; index < this->MaxThreadNum; index++) {//找空位
        if(this->Threads[index] <= 0) break;
    }

    if(index >= this->MaxThreadNum) return TP_FAIL;//未找到空位

    //创建线程
    if(pthread_create(&(this->Threads[index]), NULL, routine, (void *)this) == 0) {
        this->ActiveThreadNum++;
        return TP_SUCCESS;//添加成功
    }

    return TP_INSERT_FAIL;//添加失败
}

//================================================================== 
//Description: Init ThreadPool
//Calls: ThreadPool_Insert
//Calls By: 
//Input：
//      MaxThreads    :   Max Work Thread in the ThreadPool.
//      ActiveThreads :   Active Work Thread in the ThreadPool
//Return: handler of the ThreadPool.
//==================================================================
ThreadPool_Handler ThreadPool_Init(int MaxThreads, int ActiveThreads)
{
    ThreadPool_t *this = NULL;

    if(MAXTHREADNUM < MaxThreads) MaxThreads = MAXTHREADNUM;
    if(MaxThreads < ActiveThreads) ActiveThreads = MaxThreads;

    this = (ThreadPool_t *)malloc(sizeof(ThreadPool_t));
    if(NULL == this) return NULL;
    memset(this, 0, sizeof(ThreadPool_t));

    //销毁开关
    this->ShutDown = 0;

    //初始化线程锁
    pthread_mutex_init(&(this->Mutex), NULL);
    pthread_mutex_lock(&(this->Mutex));

    //分配线程数组
    this->Threads = (pthread_t *)malloc(sizeof(pthread_t) * MaxThreads);
    if(NULL == this->Threads) {
        pthread_mutex_unlock(&(this->Mutex));
        pthread_mutex_destroy(&(this->Mutex));
        free(this);
        return NULL;
    }
    memset(this->Threads, 0, sizeof(pthread_t) * MaxThreads);

    //初始化条件变量
    pthread_cond_init(&(this->Cond), NULL);

    //任务链表
    this->TaskNum = 0;
    this->TaskHead = &(this->_Task);//使用这种方法不需要额外申请内存空间

    //线程组
    this->MaxThreadNum = MaxThreads;//最大线程数
    this->TotalThreadNum = ActiveThreads;//目标线程数
    this->ActiveThreadNum = 0;//当前实际运行中的线程数
    this->FreeThreadNum = 0;//当前空闲线程数

    //启动线程
    while (this->ActiveThreadNum < this->TotalThreadNum) {
        ThreadPool_Insert(this);
    }

    //解锁,线程开始运行
    pthread_mutex_unlock(&(this->Mutex));

    printf("---- THREADPOOL START ----\r\n");

    return (ThreadPool_Handler)this;
}

//================================================================== 
//Description: Destory ThreadPool
//Calls: 
//Calls By: 
//Input：
//      handler :   handler of the ThreadPool.
//Return:
//      TP_SUCCESS
//      TP_NULL_POINT
//==================================================================
int ThreadPool_Destory(ThreadPool_Handler handler)
{
    int index;
    ThreadPool_t *this = (ThreadPool_t *)handler;
    if(NULL == this) return TP_NULL_POINT;

    //退出开关打开
    pthread_mutex_lock(&(this->Mutex));
    this->ShutDown = 1;
    pthread_mutex_unlock(&(this->Mutex));

    //打开所有被条件变量阻塞的线程
	pthread_cond_broadcast(&(this->Cond));

    //依次回收所有线程
    for(index = 0;index < this->MaxThreadNum;index++) {
        if(this->Threads[index] <= 0) continue;
        pthread_join(this->Threads[index], NULL);//回收线程
        printf("JOIN %ld...\r\n", this->Threads[index]);
    }

    pthread_cond_destroy(&(this->Cond));//销毁条件变量
    pthread_mutex_destroy(&(this->Mutex));//销毁线程锁

    free(this->Threads);//释放线程数组
    free(this);//释放句柄

    printf("---- THREADPOOL STOP  ----\r\n");

    return TP_SUCCESS;
}

//================================================================== 
//Description: Add a Task to ThreadPool at list tail.
//Calls: TaskNode_Init/TaskNode_Push_Tail/TaskNode_Destory
//Calls By: 
//Input：
//      handler :   handler of the ThreadPool.
//      task    :   task function.
//      arg     :   arg of the task function.
//Return:
//      TP_SUCCESS
//      TP_NULL_POINT
//      TP_TASK_INIT_FAIL
//      TP_TASK_INSERT_FAIL
//==================================================================
int ThreadPool_Add_Task_Tail(ThreadPool_Handler handler, TaskFunc task, void *arg)
{
    int ret;
    TaskNode *node = NULL;
    ThreadPool_t *this = (ThreadPool_t *)handler;
    if(NULL == this || NULL == task) return TP_NULL_POINT;

    node = TaskNode_Init(task, arg);
    if(NULL == node) return TP_TASK_INIT_FAIL;

    pthread_mutex_lock(&(this->Mutex));//上锁

    //添加任务
    ret = TaskNode_Push_Tail(this->TaskHead, node);
    if(ret == 0) {//添加成功
        this->TaskNum++;
        ret = TP_SUCCESS;
    } else {
        printf("ADD TASK FAIL\r\n");
        ret = TP_TASK_INSERT_FAIL;
    }

    pthread_mutex_unlock(&(this->Mutex));//解锁

    if(ret == TP_SUCCESS) {
        pthread_cond_signal(&(this->Cond));//条件变量唤醒
    } else {
        TaskNode_Destory(node);
    }

    return ret;
}

//================================================================== 
//Description: Add a Task to ThreadPool at list front.
//Calls: TaskNode_Init/TaskNode_Push_Front/TaskNode_Destory
//Calls By: 
//Input：
//      handler :   handler of the ThreadPool.
//      task    :   task function.
//      arg     :   arg of the task function.
//Return:
//      TP_SUCCESS
//      TP_NULL_POINT
//      TP_TASK_INIT_FAIL
//      TP_TASK_INSERT_FAIL
//==================================================================
int ThreadPool_Add_Task_Front(ThreadPool_Handler handler, TaskFunc task, void *arg)
{
    int ret = 0;
    TaskNode *node = NULL;

    ThreadPool_t *this = (ThreadPool_t *)handler;
    if(NULL == this || NULL == task) return TP_NULL_POINT;

    node = TaskNode_Init(task, arg);
    if(NULL == node) return TP_TASK_INIT_FAIL;

    pthread_mutex_lock(&(this->Mutex));//上锁

    //添加任务
    ret = TaskNode_Push_Front(this->TaskHead, node);
    if(ret == 0) {//添加成功
        this->TaskNum++;
        ret = TP_SUCCESS;
    } else {
        printf("ADD TASK FAIL\r\n");
        ret = TP_TASK_INSERT_FAIL;
    }

    pthread_mutex_unlock(&(this->Mutex));//解锁

    if(ret == TP_SUCCESS) {
        pthread_cond_signal(&(this->Cond));//条件变量唤醒
    } else {
        TaskNode_Destory(node);
    }
    
    return ret;
}

//================================================================== 
//Description: Insert (num) Work Threads into ThreadPool.
//Calls: ThreadPool_Insert
//Calls By: 
//Input：
//      handler :   handler of the ThreadPool.
//      num     :   insert num.
//Return:
//      TP_SUCCESS
//      TP_NULL_POINT
//==================================================================
int ThreadPool_Insert_Threads(ThreadPool_Handler handler, const int num)
{
    int InsertNum = 0;
    ThreadPool_t *this = (ThreadPool_t *)handler;
    if(0 >= num) return TP_SUCCESS;
    if(NULL == this) return TP_NULL_POINT;

    pthread_mutex_lock(&(this->Mutex));//上锁

    this->TotalThreadNum += num;//获取添加后的线程数
    if(this->TotalThreadNum >= this->MaxThreadNum) this->TotalThreadNum = this->MaxThreadNum;//最大线程数

    //启动线程
    while (this->ActiveThreadNum < this->TotalThreadNum) {
        ThreadPool_Insert(this);
    }

    pthread_mutex_unlock(&(this->Mutex));//解锁

    return TP_SUCCESS;
}

//================================================================== 
//Description: Remove (num) Work Threads from ThreadPool.
//Calls: 
//Calls By: 
//Input：
//      handler :   handler of the ThreadPool.
//      num     :   remove num.
//Return:
//      TP_SUCCESS
//      TP_NULL_POINT
//==================================================================
int ThreadPool_Remove_Threads(ThreadPool_Handler handler, const int num)
{
    ThreadPool_t *this = (ThreadPool_t *)handler;
    if(0 >= num) return TP_SUCCESS;
    if(NULL == this) return TP_NULL_POINT;

    pthread_mutex_lock(&(this->Mutex));//上锁

    this->TotalThreadNum -= num;//移除后的线程数
    if(this->TotalThreadNum <= 0) this->TotalThreadNum = 1;//至少要有一条工作线程

    pthread_mutex_unlock(&(this->Mutex));//解锁
    pthread_cond_signal(&(this->Cond));//条件变量唤醒

    return TP_SUCCESS;
}

//================================================================== 
//Description: Return TotalThread Num
//Calls: 
//Calls By: 
//Input：
//      handler :   handler of the ThreadPool.
//Return:
//      total thread num.
//==================================================================
int ThreadPool_TotalThreads(ThreadPool_Handler handler)
{
    int ret = 0;
    ThreadPool_t *this = (ThreadPool_t *)handler;
    if(NULL == this) return TP_NULL_POINT;

    pthread_mutex_lock(&(this->Mutex));//上锁
    ret = this->TotalThreadNum;
    pthread_mutex_unlock(&(this->Mutex));//解锁

    return ret;
}

//================================================================== 
//Description: Return ActiveThread Num
//Calls: 
//Calls By: 
//Input：
//      handler :   handler of the ThreadPool.
//Return:
//      active thread num.
//==================================================================
int ThreadPool_ActiveThreads(ThreadPool_Handler handler)
{
    int ret = 0;
    ThreadPool_t *this = (ThreadPool_t *)handler;
    if(NULL == this) return TP_NULL_POINT;

    pthread_mutex_lock(&(this->Mutex));//上锁
    ret = this->ActiveThreadNum;
    pthread_mutex_unlock(&(this->Mutex));//解锁

    return ret;
}

//================================================================== 
//Description: Return FreeThread Num
//Calls: 
//Calls By: 
//Input：
//      handler :   handler of the ThreadPool.
//Return:
//      free thread num.
//==================================================================
int ThreadPool_FreeThreads(ThreadPool_Handler handler)
{
    int ret = 0;
    ThreadPool_t *this = (ThreadPool_t *)handler;
    if(NULL == this) return TP_NULL_POINT;

    pthread_mutex_lock(&(this->Mutex));//上锁
    ret = this->FreeThreadNum;//空闲线程数
    pthread_mutex_unlock(&(this->Mutex));//解锁

    return ret;
}

//================================================================== 
//Description: Return Task Num
//Calls: 
//Calls By: 
//Input：
//      handler :   handler of the ThreadPool.
//Return:
//      task num.
//==================================================================
int ThreadPool_TaskNum(ThreadPool_Handler handler)
{
    int ret = 0;
    ThreadPool_t *this = (ThreadPool_t *)handler;
    if(NULL == this) return TP_NULL_POINT;

    pthread_mutex_lock(&(this->Mutex));//上锁
    ret = this->TaskNum;//任务数
    pthread_mutex_unlock(&(this->Mutex));//解锁

    return ret;
}

//================================================================== 
//Description: Return ThreadPool is closed
//Calls: 
//Calls By: 
//Input：
//      handler :   handler of the ThreadPool.
//Return:
//      0 running, 1 shutdown.
//==================================================================
int ThreadPool_ShutDown(ThreadPool_Handler handler)
{
    int ret = 0;
    ThreadPool_t *this = (ThreadPool_t *)handler;
    if(NULL == this) return TP_NULL_POINT;

    pthread_mutex_lock(&(this->Mutex));//上锁
    ret = this->ShutDown;//关闭标志位
    pthread_mutex_unlock(&(this->Mutex));//解锁

    return ret;
}

